package com.raymond.client.session;

import com.raymond.client.config.ClientSocketConfig;
import com.raymond.client.connector.ExponentialBackOffRetry;
import com.raymond.client.connector.RetryPolicy;
import com.raymond.client.msg.BaseMessage;
import com.raymond.client.msg.MessageManages;
import com.raymond.client.remoting.RemotingHelper;
import com.raymond.client.service.Callback;
import com.raymond.client.utils.CachedMillisecondClock;
import io.netty.channel.*;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 客户端session
 *
 * @author :  raymond
 * @version :  V1.0
 * @date :  2022-10-27 14:40
 */
public abstract class AbstractClientSession implements ChannelSession {
    private static final Logger logger = LoggerFactory.getLogger(AbstractClientSession.class);
    /**
     * session 当前状态
     * 0 初始状态
     * 1 连接状态
     * 2 注销状态
     */
    protected int state;

    /**
     * 登录时间
     */
    protected final AtomicLong connectTime = new AtomicLong();
    protected ChannelHandlerContext ctx;

    protected Channel channel;

    protected final SessionHandler sessionHandler;

    /**
     * 消息管理器
     */
    protected final WindowManager msgManager;
    protected final SessionManager sessionManager;
    /** 端口号 **/
    protected int localPort;

    protected final ClientSocketConfig socketConfig;

    protected RetryPolicy retryPolicy;

    protected boolean shutdown;

    private AsyncTaskHandlerThread asyncTaskHandlerThread;

    private final BlockingQueue<MessageManages> asyncTask = new LinkedBlockingQueue<>();

    public AbstractClientSession(SessionManager sessionManager, ClientSocketConfig socketConfig) {
        this(sessionManager, socketConfig, null);
    }

    public AbstractClientSession(SessionManager sessionManager, ClientSocketConfig socketConfig, RetryPolicy retryPolicy) {
        this.socketConfig = socketConfig;
        this.sessionManager = sessionManager;
        this.msgManager = new WindowManager(socketConfig.getWindowSize());
        sessionHandler = new SessionHandler();
        if (retryPolicy == null) {
            this.retryPolicy = new ExponentialBackOffRetry(1000, Integer.MAX_VALUE, 60 * 1000);
        }
    }

    public synchronized void sync() throws InterruptedException {
        if (isOpen()) {
            return;
        }
        wait();
    }

    @Override
    public boolean isOpen() {
        return state == 1;
    }

    @Override
    public boolean isShutdown() {
        return shutdown;
    }

    public boolean isActive() {
        return channel.isActive();
    }

    public int getWinSize() {
        return msgManager.size();
    }

    @Override
    public int getState() {
        return state;
    }

    @Override
    public Channel getChannel() {
        return ctx.channel();
    }

    @Override
    public synchronized void shutdown() throws InterruptedException {
        if (this.shutdown) {
            return;
        }
        this.shutdown = true;
        this.state = 2;
        final String addrRemote = RemotingHelper.parseChannelRemoteAddr(channel);
        channel.close().addListener((ChannelFutureListener) future ->
                logger.info("closeChannel: close the connection to remote address[{}] result: {}", addrRemote,
                future.isSuccess())).sync();
        sessionManager.shutdown(this);
    }

    public synchronized void connectSuc(ChannelHandlerContext ctx) {
        state = 1;
        connectTime.set(CachedMillisecondClock.INS.now());
        this.ctx = ctx;
        this.channel = ctx.channel();
        InetSocketAddress socket = (InetSocketAddress) this.channel.localAddress();
        AbstractClientSession.this.localPort = socket.getPort();
        fireChannelActive(this.channel);
        asyncTaskHandlerThread = new AsyncTaskHandlerThread();
        asyncTaskHandlerThread.start();
        notifyAll();
    }

    public MessageManages syncSend(BaseMessage req) {
        MessageManages messageManages = new MessageManages(req);
        msgManager.put(req.getSequenceId(), messageManages, socketConfig.getRequestExpiryTimeout());
        channel.writeAndFlush(req);
        messageManages.wait0(socketConfig.getResponseExpiryTimeout());
        if (messageManages.isDone()) {
            return messageManages;
        }
        msgManager.remove(req.getSequenceId());
        return messageManages.setCode(-2);
    }

    public void asyncSend(BaseMessage req, Callback callback) {
        asyncSend(new MessageManages(req, callback));
    }

    public void asyncSend(MessageManages messageManages) {
        msgManager.put(messageManages.getReq().getSequenceId(), messageManages, socketConfig.getRequestExpiryTimeout());
        channel.writeAndFlush(messageManages.getReq()).addListener(future -> {
            if (future.isSuccess()) {
                asyncTask.add(messageManages.setStartTime(CachedMillisecondClock.INS.now()));
            } else if (messageManages.isRetry()) {
                asyncSend(messageManages);
            } else {
                messageManages.asyncSendFail();
            }
        });

    }



    /**
     * 连接成功后初始化 channel
     * @param channel channel
     */
    protected abstract void fireChannelActive(Channel channel);

    public SessionHandler getSessionHandler() {
        return sessionHandler;
    }


    public ClientSocketConfig getSocketConfig() {
        return socketConfig;
    }

    public AbstractClientSession connect() {
        if (isOpen()) {
            return this;
        }
        sessionManager.getConnector().bind(this);
        return this;
    }

    /**
     * 心跳请求对象
     * @return 心跳请求对象
     */
    public Object getHeartbeatRequest() {
        return null;
    }



    class AsyncTaskHandlerThread extends Thread {

        @Override
        public void run() {
            setName("AsyncTaskHandlerThread");
            while (!shutdown) {
                try {
                    execute();
                } catch (Exception e) {
                    logger.error("处理异步任务异常,{}", getName(), e);
                }
            }
        }

        private void execute() {
            try {
                MessageManages take = asyncTask.poll(3, TimeUnit.SECONDS);
                if (take == null || take.isDone()) {
                    return;
                }
                long sleepTime = (take.getStartTime() + socketConfig.getResponseExpiryTimeout()) - CachedMillisecondClock.INS.now();
                if (sleepTime > 0) {
                    take.wait0(sleepTime);
                }
                if (sleepTime < 0) {
                    logger.debug("异步任务剩余数据：" + asyncTask.size() + ",等待时间:" + sleepTime +
                            ",开始时间:" + take.getStartTime() + ",当前时间:" + CachedMillisecondClock.INS.now() +
                            ",getSequenceId:" + take.getReq().getSequenceId());
                }

                if (!take.isAsyncDone()) {
                    take.setCode(-2);
                    take.asyncSendFail();
                    AbstractClientSession.this.msgManager.remove(take.getReq().getSequenceId());
                }
            } catch (Exception e) {
                logger.error("处理异步线程异常:", e);
            }
        }


    }

    @ChannelHandler.Sharable
    class SessionHandler extends ChannelInboundHandlerAdapter {

        /** 当前重试次数 **/
        private int retries = 0;

        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            logger.debug("Successfully established a connection to the server. remoteAddress:{}:{}.",
                    socketConfig.getHost(), socketConfig.getPort());
            retries = 0;
            connectSuc(ctx);
            ctx.fireChannelActive();
        }

        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            state = 2;
            if (retries == 0) {
                logger.debug("Lost the TCP connection with the server. remoteAddress:{}:{}. localPort:{}",
                        socketConfig.getHost(), socketConfig.getPort(), localPort);
                ctx.close();
            }
            boolean allowRetry = retryPolicy.allowRetry(retries);
            if (allowRetry && !isShutdown()) {
                long sleepTimeMs = retryPolicy.getSleepTimeMs(retries);
                logger.debug("Try to reconnect to the server after {}ms. Retry count: {}.", sleepTimeMs, ++retries);
                final EventLoop eventLoop = ctx.channel().eventLoop();
                eventLoop.schedule(() -> {
                    logger.debug("Reconnecting remoteAddress {}:{} ... ",
                            socketConfig.getHost(), socketConfig.getPort());
                    connect();
                }, sleepTimeMs, TimeUnit.MILLISECONDS);
            }
            ctx.fireChannelInactive();
        }

        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
            if (evt instanceof IdleStateEvent) {
                IdleState state = ((IdleStateEvent) evt).state();
                if (state == IdleState.ALL_IDLE) {
                    final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
                    logger.debug("发送心跳,remoteAddress:{},channelId:{},localPort:{}", remoteAddress, ctx.channel().id(), localPort);
                    ctx.writeAndFlush(getHeartbeatRequest());
                }
            } else {
                super.userEventTriggered(ctx, evt);
            }
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            super.exceptionCaught(ctx, cause);
        }
    }
}
